package com.example.admin.service.impl;

import com.alibaba.fastjson.JSON;
import com.example.admin.mapper.PersonMapper;
import com.example.admin.model.Person;
import com.example.admin.result.ResponseResult;
import com.example.admin.service.ExampleService;
import io.seata.spring.annotation.GlobalTransactional;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

@Service
public class ExampleServiceImpl implements ExampleService {

    private static final Logger logger = LoggerFactory.getLogger(ExampleServiceImpl.class);

    private static int num = 0;

    @Value("${spring.kafka.template.default-topic}")
    private String kafkatopic;

    @Value("${redis.topic}")
    private String redistopic;

    @Autowired
    KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Resource
    PersonMapper personMapper;

    @Override
    public ResponseResult<String> producerKafkaExample() {
        ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send(kafkatopic, "key0001" + num, "value0001" + num);
        logger.info("send: {}", JSON.toJSON(send));
        return ResponseResult.succeed("00000", "HELLO WORLD");
    }

    @Override
    public ResponseResult<String> consumerKafkaExample(ConsumerRecord consumerRecord) {
        logger.info(String.format("topic:%s,分区：%d,偏移量：%d," + "key:%s,value:%s" + "计数器:%s",consumerRecord.topic(),consumerRecord.partition(),consumerRecord.offset(),consumerRecord.key(),consumerRecord.value(), num++));
        return ResponseResult.succeed("00000", "HELLO WORLD");
    }

    /**
     * RedisTemplate使用的是 JdkSerializationRedisSerializer 序列化对象
     * StringRedisTemplate使用的是 StringRedisSerializer 序列化String
     * @return
     */
    @Override
    public ResponseResult<String> producerRedisExample(List<Person> personList) {
        Map<String, Object> map = personList.stream().collect(Collectors.toMap(Person::getCustID, Function.identity()));
        stringRedisTemplate.convertAndSend(redistopic, JSON.toJSONString(map));
        return ResponseResult.succeed("00000", "HELLO WORLD");
    }

    @Override
    public ResponseResult<String> saveSeataAndRedisExample(List<Person> personList) {
        stringRedisTemplate.setEnableTransactionSupport(true);
        stringRedisTemplate.multi();
        int a = 0;
        stringRedisTemplate.opsForValue().set("骆永鹏", JSON.toJSONString(personList));
        logger.info("saveSeataAndRedisExample: {}", 1/0);
        stringRedisTemplate.opsForValue().set("刘银银", JSON.toJSONString(personList));
        stringRedisTemplate.exec();
        return ResponseResult.succeed("00000", "HELLO WORLD");
    }

    @Override
    @GlobalTransactional
    public ResponseResult<String> savePersonExample(List<Person> personList) {
        int a = 0;
        personMapper.savePersonExample(personList);
//        logger.info("saveSeataAndRedisExample: {}", 1/0);
        personMapper.savePersonExample(personList);
        return ResponseResult.succeed("00000", "HELLO WORLD");
    }

    @Override
    public ResponseResult<List<Person>> queryPersonExample() {
        List<Person> list = personMapper.queryPersonExample();
        return ResponseResult.succeed("00000", "操作成功", list, list.size());
    }

}